package com.zyk.scaffold.mq.rocket.business.order.consumers;

import com.alibaba.fastjson.JSONObject;
import com.zyk.scaffold.mq.rocket.business.order.service.OrderService;
import com.zyk.scaffold.mq.rocket.globs.constants.OrderStatus;
import com.zyk.scaffold.mq.rocket.repositorys.interfaces.TOrderRepository;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.*;
import org.apache.rocketmq.client.apis.message.MessageId;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.io.IOException;
import java.util.Collections;
import java.util.Map;

@Slf4j
@Component
public class OrderConsumer {

    private PushConsumer pushConsumer;

    @Autowired
    private TOrderRepository orderRepository;

    @Autowired
    private OrderService orderService;

    public OrderConsumer() throws ClientException, IOException, InterruptedException {
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        // 接入点地址，需要设置成Proxy的地址和端口列表，一般是xxx:8081;xxx:8081。
        String endpoints = "localhost:8081";
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoints)
                .build();
        // 订阅消息的过滤规则，表示订阅所有Tag的消息。
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // 为消费者指定所属的消费者分组，Group需要提前创建。
        String consumerGroup = "YourConsumerGroup";
        // 指定需要订阅哪个目标Topic，Topic需要提前创建。
        String topic = "TestTopic";
        // 初始化PushConsumer，需要绑定消费者分组ConsumerGroup、通信参数以及订阅关系。
        pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                // 设置消费者分组。
                .setConsumerGroup(consumerGroup)
                // 设置预绑定的订阅关系。
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                // 设置消费监听器。
                .setMessageListener(new MessageListenerConcurrently4Pay())
                .build();
    }
    class MessageListenerConcurrently4Pay implements MessageListener {
        @Override
        public ConsumeResult consume(MessageView messageView) {

            try {
                String topic = messageView.getTopic();
                String msgBody = new String(messageView.getBody().array(), "utf-8");
                String tags = messageView.getTag().orElse(null);
                String keys = messageView.getKeys().stream().findFirst().orElse(null);
                System.err.println("收到消息：" + "  topic :" + topic + "  ,tags : " + tags + "keys :" + keys + ", msg : " + msgBody);
                MessageId orignMsgId = messageView.getMessageId();
                System.err.println("orignMsgId: " + orignMsgId);

                //通过keys 进行去重表去重 或者使用redis进行去重???? --> 不需要

                Map<String, Object> body = JSONObject.parseObject(msgBody, Map.class);
                String orderId = (String) body.get("orderId");
                String userId = (String) body.get("userId");
                String status = (String)body.get("status");


                if(status.equals(OrderStatus.ORDER_PAYED.getValue())) {
                    int count  = orderRepository.updateOrderStatus(orderId, status, "admin");
                    if(count == 1) {
                        orderService.sendOrderlyMessage4Pkg(userId, orderId);
                    }

                }
            } catch (Exception e) {
                e.printStackTrace();
                return ConsumeResult.FAILURE;
            }
            return ConsumeResult.SUCCESS;
        }
    }

    @PreDestroy
    private void destory(){
        if(pushConsumer != null){
            try {
                // 如果不需要再使用 PushConsumer，可关闭该实例。
                pushConsumer.close();
            } catch (IOException e) {
                log.error("pushConsumer close error", e);
            }
        }
    }

}